package com.zhilun;

import org.apache.pulsar.client.api.*;

import java.text.MessageFormat;
import java.util.concurrent.TimeUnit;

public class ZhiLunMessageConsumerPro {




    public static void main(String[] args) throws Exception {

        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://api-pulsar.zhilun.com:6650")
                .authentication(AuthenticationFactory.token(" eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJkY2R6UHJvZCJ9.THguiaThFLFY9PI2vmyo696DdwCiXsRfH9F4ZxxbspRF46U8-7tarsDuuXB0TsO40FsTwiVLSyzUPeNeChRMmnjQZzYYg_GBaIq84tp33WI3ScsSFLEX-mVV2MeYAybcUQQ7KWa3vwecXiioRU4PC2SgZGwQhX3kfLrvaJd9flvhrFhhnxoChzv0xHXwL-yhMaKWL1ZcO8ync-_WpPB8VGnJ7ChPnE3IMTettZ4rHC5sFOuWCkY04sPu4ELNyw3YCFoo_VZ6IH4MYiwO0DeMd0hyFOtECMZE7vgABK2m3gJsDmuKDL6C1CfLLLhxv12tpMl4ula18j7_kQohls5K1w"))
                .build();
        //初始化消费者  订阅alarmEvent Topic
         client.newConsumer()
                .topic("non-persistent://companyAppId78/msg/carrier","non-persistent://companyAppId78/msg/tpms")
                .subscriptionName("dcdz-listener-group")
                .subscriptionType(SubscriptionType.Shared)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪里开始消费还有Latest，valueof可选，默认Latest
                .negativeAckRedeliveryDelay(10, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .messageListener((MessageListener<byte[]>) (consumer, message) -> {
                    //接收message处理程序
                    System.out.println(MessageFormat.format("接收pulsar,topic:{0},key:{1},value:{2}",message.getTopicName(), message.getKey() ,new String(message.getData())));

                })
                .subscribe();

    }
}


